package com.yihaodian.indices;

import java.util.Collection;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;

import org.apache.log4j.Logger;

import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.common.collect.Sets.SetView;
import com.google.inject.Inject;
import com.yihaodian.api.search.SearchException;
import com.yihaodian.cluster.ClusterChangedEvent;
import com.yihaodian.cluster.ClusterService;
import com.yihaodian.cluster.ClusterState;
import com.yihaodian.cluster.ClusterStateListener;
import com.yihaodian.cluster.Node;
import com.yihaodian.index.IndexShard;
import com.yihaodian.index.data.IndexDataService;
import com.yihaodian.index.meta.IndexMeta;
import com.yihaodian.index.meta.shard.ShardId;
import com.yihaodian.index.meta.shard.ShardRouting;
import com.yihaodian.search.broker.ThreadPools;
import com.yihaodian.server.LifeCycleService;

public class IndicesClusterStateService implements ClusterStateListener,
		LifeCycleService {
	static Logger logger = Logger.getLogger(IndicesClusterStateService.class);

	ClusterService clusterService;
	IndicesService indicesService;
	IndexDataService indexDataService;

	@Inject
	public IndicesClusterStateService(ClusterService clusterService,
			IndicesService indicesService, IndexDataService indexDataService) {
		this.clusterService = clusterService;
		this.indicesService = indicesService;
		this.indexDataService = indexDataService;
	}

	@Override
	public void onClusterState(ClusterChangedEvent event) {
		final ClusterState current = event.state();

		Node node = clusterService.getLocalNode();
		final Collection<ShardRouting> shards = current.routingTables()
				.getShardRouting(node.getId());

		ExecutorService es = ThreadPools.getThreadPool(ThreadPools.COMMON);
		es.submit(new Runnable() {

			public void run() {
				try {
					manageShard(current, shards);
				} catch (Exception e) {
					logger.error(e, e);
				}
			}
		});
	}

	synchronized void manageShard(ClusterState cs,
			Collection<ShardRouting> shardRouting) {
		Set<ShardRouting> currentSet = new HashSet<ShardRouting>(
				indicesService.shardRouting());
		Set<ShardRouting> targetSet = new HashSet<ShardRouting>(shardRouting);

		SetView<ShardRouting> current = Sets
				.intersection(currentSet, targetSet);
		SetView<ShardRouting> remove = Sets.difference(currentSet, targetSet);
		SetView<ShardRouting> add = Sets.difference(targetSet, currentSet);

		// manage enable/disable

		boolean routingChange = false;
		if (current.size() > 0) {
			Set<ShardId> shardIds = Sets.newHashSet();
			for (ShardRouting sr : current) {
				shardIds.add(sr.getShardId());
			}

			currentShard(currentSet, targetSet, shardIds, cs);
		}

		if (add.size() > 0) {
			addShard(cs, add);
			routingChange = true;
		}

		if (remove.size() > 0) {
			removeShard(remove);
			routingChange = true;
		}

		if (routingChange) {
			// TODO
			// this.clusterService.updateClusterState(update);
		}
	}

	void currentShard(Set<ShardRouting> currentSet,
			Set<ShardRouting> targetSet, Set<ShardId> shardIds, ClusterState cs) {
		logger.info("current shard   " + currentSet);
		logger.info("target shard  " + targetSet);

		Map<ShardId, ShardRouting> currentMap = asMap(currentSet);
		Map<ShardId, ShardRouting> targetMap = asMap(targetSet);
		for (final ShardId shardId : shardIds) {
			ShardRouting current = currentMap.get(shardId);
			final ShardRouting target = targetMap.get(shardId);

			if (current.isEnable() != target.isEnable()) {
				if (target.isEnable()) {
					indicesService.enableShard(shardId);
				} else {
					indicesService.disableShard(shardId);
				}
			}

			IndexMeta indexMeta = cs.indexMetas().indexMeta(
					shardId.getIndexId());
			final String version = indexMeta.version(shardId);
			if (!current.getVersion().equals(version)) {
				ExecutorService es = ThreadPools.getThreadPool(ThreadPools.COMMON);
				es.submit(new Runnable() {

					public void run() {
						try {
							IndexShard indexShard = indicesService.getIndexShard(shardId);
							indexDataService.download(target.getShardId(), target.getVersion());
							indexShard.switchIndex(target.getVersion());

							// TODO update to ZK
							target.setVersion(version);
						} catch (Exception e) {
							logger.error(e, e);
						}
					}
				});
			}
		}
	}

	Map<ShardId, ShardRouting> asMap(Set<ShardRouting> set) {
		Map<ShardId, ShardRouting> map = Maps.newHashMap();
		for (ShardRouting s : set)
			map.put(s.getShardId(), s);

		return map;
	}

	void addShard(ClusterState cs, SetView<ShardRouting> shards) {
		logger.info("add shard" + shards);
		for (ShardRouting shard : shards) {
			IndexMeta indexMeta = cs.indexMetas().indexMeta(
					shard.getShardId().getIndexId());
			indicesService.startIndexShard(indexMeta, shard);

			// TODO, update to ZK
			shard.setStatus(ShardRouting.Status.START);
			shard.setVersion(indexMeta.version(shard.getShardId()));
		}
	}

	void removeShard(SetView<ShardRouting> shards) {
		logger.info("remove shard" + shards);
		for (ShardRouting shard : shards) {
			indicesService.stopIndexShard(shard.getShardId());
		}
	}

	@Override
	public void startup() throws SearchException {
		this.clusterService.addClusterStateListener(this);
	}

	@Override
	public void shutdown() throws SearchException {

	}

}
